package net.bwie.realtime.jtp.dwd.trade.job;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.jtp.common.utils.DateTimeUtil;
import net.bwie.realtime.jtp.common.utils.KafkaUtil;
import net.bwie.realtime.jtp.dwd.trade.function.RequestDicDimAsyncFunction;
import net.bwie.realtime.jtp.dwd.trade.function.LoadProvinceDimMapFunction;
import net.bwie.realtime.jtp.dwd.trade.util.DorisUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * @Author: FuHe
 * @Date: 2025/6/3
 */
public class JtpOrderDetailDorisDwdJob {
    public static void main(String[] args) throws Exception {
        // 1 执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 2 数据源 kafka
        DataStream<String> dataStream = KafkaUtil.consumerKafka(env, "topic-db");
        dataStream.print();
        // 3 数据转换
        DataStream<String> dwdStream = handle(dataStream);
//        dwdStream.print();
        // 4 数据接收器
        DorisUtil.saveToDoris(dwdStream, "jtp_mall_doris", "dwd_order_detail");
        // 5 触发执行
        env.execute("JtpOrderDetailDorisDwdJob");
    }

    private static DataStream<String> handle(DataStream<String> dataStream) {
        // 1 数据库JOIN：Interval Join，订单明细数据流关联订单信息数据流
        DataStream<String> joinStream = joinProcess(dataStream);
//        joinStream.print();
        // 2 省份维度补全：异步io
        DataStream<String> provinceStream = joinStream.map(new LoadProvinceDimMapFunction());
//        provinceStream.print();
        // 3 字典维度补全：异步IO
        DataStream<String> dwdStream = AsyncDataStream.orderedWait(
                provinceStream,
                new RequestDicDimAsyncFunction(),
                10000,
                TimeUnit.MILLISECONDS,
                100);
        return dwdStream;
    }

    private static DataStream<String> joinProcess(DataStream<String> stream) {
        // 1.1 订单明细数据流
        SingleOutputStreamOperator<String> detailStream = stream.filter(
                new RichFilterFunction<String>() {
                    @Override
                    public boolean filter(String value) throws Exception {
                        // 解析json数据
                        JSONObject jsonObject = JSON.parseObject(value);
                        // 过滤
                        return "order_detail".equals(jsonObject.getString("table_name"))
                                && "insert".equals(jsonObject.getString("operate_type"));
                    }
                }
        );
        // 1-2  订单数据流
        SingleOutputStreamOperator<String> infoStream = stream.filter(
                new RichFilterFunction<String>() {
                    @Override
                    public boolean filter(String value) throws Exception {
                        // 解析json数据
                        JSONObject jsonObject = JSON.parseObject(value);
                        // 过滤
                        return "order_info".equals(jsonObject.getString("table_name"))
                                && "insert".equals(jsonObject.getString("operate_type"));
                    }
                }
        );
        // 2-1 订单数据流
        SingleOutputStreamOperator<String> timeInfoStream = infoStream.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<String>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<String>() {
                                    @Override
                                    public long extractTimestamp(String element, long recordTimestamp) {
                                        // 解析
                                        String eventTime = JSON.parseObject(element)
                                                .getJSONObject("operate_data").getString("create_time");
                                        // 转换时间戳
                                        return DateTimeUtil.convertStringToLong(eventTime, "yyyy-MM-dd HH:mm:ss");
                                    }
                                }
                        )
        );
        // 2-2). 订单明细流
        SingleOutputStreamOperator<String> timeDetailStream = detailStream.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<String>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<String>() {
                                    @Override
                                    public long extractTimestamp(String element, long recordTimestamp) {
                                        // 解析
                                        String eventTime = JSON.parseObject(element)
                                                .getJSONObject("operate_data").getString("create_time");
                                        // 转换时间戳
                                        return DateTimeUtil.convertStringToLong(eventTime, "yyyy-MM-dd HH:mm:ss");
                                    }
                                }
                        )
        );
        // 3-1 info和detail
        SingleOutputStreamOperator<String> joinStream = timeInfoStream
                .keyBy(
                        element -> JSON.parseObject(element).getJSONObject("operate_data").getString("id")
                ).intervalJoin(timeDetailStream.keyBy(
                                element -> JSON.parseObject(element).getJSONObject("operate_data").getString("order_id")
                        )
                ).between(Time.seconds(-1), Time.seconds(2))
                .process(
                        new ProcessJoinFunction<String, String, String>() {
                            @Override
                            public void processElement(String left, String right, Context ctx, Collector<String> out) throws Exception {
                                // 获取user_id和province_id
                                JSONObject infoJsonObject = JSON.parseObject(left).getJSONObject("operate_data");
                                Integer userId = infoJsonObject.getInteger("user_id");
                                Integer provinceId = infoJsonObject.getInteger("province_id");
                                // 将字段添加到detail中
                                JSONObject detailJsonObject = JSON.parseObject(right).getJSONObject("operate_data");
                                detailJsonObject.put("user_id", userId);
                                detailJsonObject.put("province_id", provinceId);
                                // 订单日期
                                String createTime = detailJsonObject.getString("create_time");
                                String curDate = createTime.substring(0, 10);
                                detailJsonObject.put("cur_date", curDate);
                                // 输出
                                String output = detailJsonObject.toJSONString();
                                out.collect(output);
                            }
                        }
                );
        // 返回关联宽表数据流
        return joinStream;
    }
}
